package plus.easydo.rocketmq.config.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import plus.easydo.rocketmq.consumer.ConsumeMessageListenerProcessor;

/**
 * 消费者配置
 * @author laoyu
 * @version 1.0
 * @date 2021/11/3
 */
@Configuration
@ConfigurationProperties(prefix = "rocketmq.consumer")
public class ConsumerConfigure {

    public static final Logger LOGGER = LoggerFactory.getLogger(ConsumerConfigure.class);

    /**分组*/
    private String groupName;

    /**nameServer地址*/
    private String nameServerAddress;

    /**订阅主题集合*/
    private String topics;

    /**消费模型*/
    private String messageModel = "CLUSTERING";

    /**消费者线程最小值*/
    private Integer consumeThreadMin;

    /**消费者线程最大值*/
    private Integer consumeThreadMax;

    /**一次消费信息的条数*/
    private Integer consumeMessageBatchMaxSize;

    @Autowired
    private ConsumeMessageListenerProcessor consumeMsgListenerProcessor;

    /**
     * 消费者配置
     *
     * @return org.apache.rocketmq.client.consumer.DefaultMQPushConsumer
     * @author laoyu
     * @date 2021/11/3
     */
    @Bean
    @ConditionalOnProperty(prefix = "rocketmq.consumer", value = "enable", havingValue = "true")
    public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
        LOGGER.info("defaultConsumer 正在创建---------------------------------------");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(nameServerAddress);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        // 设置监听
        consumer.registerMessageListener(consumeMsgListenerProcessor);

        /**
         * 设置consumer第一次启动是从队列头部开始还是队列尾部开始
         * 如果不是第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        /**
         * 设置消费模型，集群还是广播，默认为集群
         */
        if(messageModel.equals(MessageModel.BROADCASTING.getModeCN())){
            consumer.setMessageModel(MessageModel.BROADCASTING);
        }

        try {
            // 设置该消费者订阅的主题和tag，如果订阅该主题下的所有tag，则使用*,
            String[] topicArr = topics.split(";");
            for (String topic : topicArr) {
                String[] tagArr = topic.split("~");
                consumer.subscribe(tagArr[0], tagArr[1]);
            }
            consumer.start();
            LOGGER.info("consumer 创建成功 groupName={}, topics={}, nameServerAddress={}",groupName,topics, nameServerAddress);
        } catch (MQClientException e) {
            LOGGER.error("consumer 创建失败!");
        }
        return consumer;
    }

    public String getGroupName() {
        return groupName;
    }

    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }

    public String getNameServerAddress() {
        return nameServerAddress;
    }

    public void setNameServerAddress(String nameServerAddress) {
        this.nameServerAddress = nameServerAddress;
    }

    public String getTopics() {
        return topics;
    }

    public void setTopics(String topics) {
        this.topics = topics;
    }

    public String getMessageModel() {
        return messageModel;
    }

    public void setMessageModel(String messageModel) {
        this.messageModel = messageModel;
    }

    public Integer getConsumeThreadMin() {
        return consumeThreadMin;
    }

    public void setConsumeThreadMin(Integer consumeThreadMin) {
        this.consumeThreadMin = consumeThreadMin;
    }

    public Integer getConsumeThreadMax() {
        return consumeThreadMax;
    }

    public void setConsumeThreadMax(Integer consumeThreadMax) {
        this.consumeThreadMax = consumeThreadMax;
    }

    public Integer getConsumeMessageBatchMaxSize() {
        return consumeMessageBatchMaxSize;
    }

    public void setConsumeMessageBatchMaxSize(Integer consumeMessageBatchMaxSize) {
        this.consumeMessageBatchMaxSize = consumeMessageBatchMaxSize;
    }

    public ConsumeMessageListenerProcessor getConsumeMsgListenerProcessor() {
        return consumeMsgListenerProcessor;
    }

    public void setConsumeMsgListenerProcessor(ConsumeMessageListenerProcessor consumeMsgListenerProcessor) {
        this.consumeMsgListenerProcessor = consumeMsgListenerProcessor;
    }
}
